package com.hna.face.mqtt;

import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;

import javax.annotation.PostConstruct;
import javax.net.ssl.SSLSocketFactory;

import org.eclipse.paho.client.mqttv3.MqttClient;
import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
import org.eclipse.paho.client.mqttv3.MqttDeliveryToken;
import org.eclipse.paho.client.mqttv3.MqttException;
import org.eclipse.paho.client.mqttv3.MqttMessage;
import org.eclipse.paho.client.mqttv3.MqttPersistenceException;
import org.eclipse.paho.client.mqttv3.MqttSecurityException;
import org.eclipse.paho.client.mqttv3.MqttTopic;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Service;

import com.hna.face.config.SysConfig;
import com.hna.face.security.ssl.SSLUtils;

@Service
public class MqttService {

	private final static Logger log = LoggerFactory.getLogger(MqttService.class);

	private ScheduledExecutorService scheduler;

	private MqttClient client;
	
	private MqttConnectOptions options;

	private MqttMessage message;
	
	@PostConstruct
	public void init() {
		try {
			client = new MqttClient(SysConfig.mqttHost, SysConfig.mqttClientid, new MemoryPersistence());
			connect();
			startReconnect();
		} catch (MqttException e1) {
			log.error("mqtt init", e1);
		}
	}

	private void connect() {
		options = new MqttConnectOptions();
		options.setCleanSession(false);
		options.setUserName(SysConfig.mqttUserName);
		options.setPassword(SysConfig.mqttPassword.toCharArray());
		// 设置超时时间
		options.setConnectionTimeout(SysConfig.mqttConnectionTimeout);
		// 设置会话心跳时间
		options.setKeepAliveInterval(SysConfig.mqttKeepAliveInterval);

		SSLUtils sslUtils = new SSLUtils();
		try {
			String sslPath = this.getClass().getClassLoader().getResource("").getFile();
			log.info(sslPath);
			
			SSLSocketFactory factory = sslUtils.getSSLSocktet("ssl/ca.crt", "ssl/client1.crt", "ssl/client1.pem", "");
			options.setSocketFactory(factory);
		} catch (Exception e1) {
			log.error("", e1);
		}

		try {
			client.setCallback(new PushCallback());
			MqttTopic topics = client.getTopic(SysConfig.mqttWill);
			// setWill方法，如果项目中需要知道客户端是否掉线可以调用该方法。设置最终端口的通知消息
			options.setWill(topics, (SysConfig.mqttClientid + " close").getBytes(), 2, true);
			client.connect(options);

			MqttMessage message = new MqttMessage();
			message.setQos(1);
			message.setRetained(true);
			message.setPayload((SysConfig.mqttClientid + SysConfig.mqttOnline).getBytes());
			publishOnline(message);
			
			subscribe(2,"xy-003");
			
			MqttMessage message1=new MqttMessage();
			message1.setRetained(true);
			message1.setQos(2);
			message1.setPayload("".getBytes());
			publish(message1,"xy-003");
			publish(message1,"data");
		} catch (Exception e) {
			log.error("connectException",e);
		}
	}
	
	public void publishOnline(MqttMessage message) throws MqttPersistenceException, MqttException {
		MqttTopic topic = client.getTopic(SysConfig.mqttOnline);
		MqttDeliveryToken token = topic.publish(message);
		token.waitForCompletion();
		log.info(token.isComplete() + "========");
	}
	

	public void publish(MqttMessage message) throws MqttPersistenceException, MqttException {
/*		MqttDeliveryToken token = topic.publish(message);
		token.waitForCompletion();
		log.info(token.isComplete() + "========");*/
	}
	
	public void publish(MqttMessage message, String topics) throws MqttPersistenceException, MqttException {
		log.info("topic"+topics);
		MqttTopic topic = client.getTopic(topics);
		MqttDeliveryToken token = topic.publish(message);
		token.waitForCompletion();
		log.info(token.isComplete() + "========");
	}

	public void subscribe() throws MqttException {
		int[] Qos = { 1 };
		String[] topic1 = { SysConfig.mqttClientid };
		client.subscribe(topic1, Qos);
		log.info("subscribe" + "========");
	}
	
	public void subscribe(int qos,String topics) throws MqttException {
		log.info("subscribeTopics " + topics+"start");
		int[] Qos = { qos };
		String[] topic1 = { topics };
		if (client == null) {
			log.error("client is null");
		}
		client.subscribe(topic1, Qos);
		log.info("subscribeTopics " + topics+"end");
	}

	// 重新链接
	public void startReconnect() {
		scheduler = Executors.newSingleThreadScheduledExecutor();
		scheduler.scheduleAtFixedRate(new Runnable() {
			public void run() {
				if (!client.isConnected()) {
					try {
						log.info("重新链接");
						client.connect(options);

						MqttMessage message = new MqttMessage();
						message.setQos(1);
						message.setRetained(true);
						message.setPayload((SysConfig.mqttClientid + SysConfig.mqttOnline).getBytes());
						publishOnline(message);

					} catch (MqttSecurityException e) {
						e.printStackTrace();
					} catch (MqttException e) {
						e.printStackTrace();
					}
				}
			}
		},SysConfig.mqttInitialDelay, SysConfig.mqttPeriod, TimeUnit.MILLISECONDS);
	}

	// 定时任务，同步信息给app
	// publish 数据到 deviceno 队列，格式：
	// 1：人员表和照片表
	// 2：排班表

//    接收上报的结果

/*	public static void main(String[] args) throws MqttException {
		MqttService server = new MqttService();
		server.message = new MqttMessage();
		server.message.setQos(1);
		server.message.setRetained(true);
		server.message.setPayload("eeeeeaaaaaawwwwww---".getBytes());
		server.publish(server.message);
		log.info(server.message.isRetained() + "------ratained状态");
	}*/

}
